package drds.server.execute_engine.data_handling;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.text.NumberFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import drds.server.Server;
import drds.server.memory.Memory;
import drds.server.memory.unsafe.KVIterator;
import drds.server.memory.unsafe.map.UnsafeFixedWidthAggregationMap;
import drds.server.memory.unsafe.memory.mm.DataNodeMemoryManager;
import drds.server.memory.unsafe.memory.mm.MemoryManager;
import drds.server.memory.unsafe.row.BufferHolder;
import drds.server.memory.unsafe.row.StructType;
import drds.server.memory.unsafe.row.UnsafeRow;
import drds.server.memory.unsafe.row.UnsafeRowWriter;
import drds.server.memory.unsafe.utils.BytesTools;
import drds.server.memory.unsafe.utils.PropertyConfig;
import drds.server.memory.unsafe.utils.sort.UnsafeExternalRowSorter;
import drds.server.util.ByteUtil;

/**
 * 
 * implement group function select a,count(*),sum(*) from A group by a
 * 
 */
public class UnsafeRowGrouper {
	private static final Logger LOGGER = LoggerFactory.getLogger(UnsafeRowGrouper.class);

	private UnsafeFixedWidthAggregationMap aggregationMap = null;
	private final Map<String, Column> columToIndx;
	private final AggregateFunctionType[] mergeColums;
	private String[] sortColumnsByIndex = null;
	private final String[] columns;
	private boolean isMergAvg = false;
	private GroupByFilter havingCols;
	private UnsafeRow groupKey = null;
	private UnsafeRow valueKey = null;
	private BufferHolder bufferHolder = null;
	private UnsafeRowWriter unsafeRowWriter = null;
	private final int groupKeyfieldCount;
	private final int valuefieldCount;
	private StructType groupKeySchema;
	private StructType aggBufferSchema;
	private UnsafeRow emptyAggregationBuffer;
	private final Memory memory;
	private final MemoryManager memoryManager;
	private final PropertyConfig propertyConfig;

	public UnsafeRowGrouper(Map<String, Column> columToIndx, String[] columns, AggregateFunctionType[] mergCols, GroupByFilter havingCols) {
		super();
		assert columns != null;
		assert columToIndx != null;
		assert mergCols != null;
		this.columToIndx = columToIndx;
		this.columns = columns;
		this.mergeColums = mergCols;
		this.havingCols = havingCols;
		this.sortColumnsByIndex = columns != null ? toSortColumnsByIndex(columns, columToIndx) : null;
		this.groupKeyfieldCount = columns != null ? columns.length : 0;
		this.valuefieldCount = columToIndx != null ? columToIndx.size() : 0;
		this.memory = Server.getInstance().getMemory();
		this.memoryManager = memory.getResultMergeMemoryManager();
		this.propertyConfig = memory.getConf();

		LOGGER.debug("columToIndx :" + (columToIndx != null ? columToIndx.toString() : "null"));

		initGroupKey();
		initEmptyValueKey();

		DataNodeMemoryManager dataNodeMemoryManager = new DataNodeMemoryManager(memoryManager, Thread.currentThread().getId());

		aggregationMap = new UnsafeFixedWidthAggregationMap(emptyAggregationBuffer, aggBufferSchema, groupKeySchema, dataNodeMemoryManager, 2 * 1024, propertyConfig.getSizeAsBytes(".buffer.pageSize", "1m"), false);
	}

	private String[] toSortColumnsByIndex(String[] columns, Map<String, Column> columToIndx) {

		Map<String, Integer> map = new HashMap<String, Integer>();

		Column colum;
		for (int i = 0; i < columns.length; i++) {
			colum = columToIndx.get(columns[i].toUpperCase());
			if (colum == null) {
				throw new IllegalArgumentException("all columns in group by clause should be in the selected column list.!" + columns[i]);
			}
			map.put(columns[i], colum.columIndex);
		}

		String[] sortColumnsByIndex = new String[map.size()];

		List<Map.Entry<String, Integer>> entryList = new ArrayList<Map.Entry<String, Integer>>(map.entrySet());

		Collections.sort(entryList, new Comparator<Map.Entry<String, Integer>>() {
			@Override
			public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {
				return o1.getValue().compareTo(o2.getValue());
			}
		});

		Iterator<Map.Entry<String, Integer>> iterator = entryList.iterator();
		Map.Entry<String, Integer> tmpEntry = null;

		int index = 0;

		while (iterator.hasNext()) {
			tmpEntry = iterator.next();
			sortColumnsByIndex[index++] = tmpEntry.getKey();
		}

		return sortColumnsByIndex;
	}

	private void initGroupKey() {
		/**
		 * 构造groupKey
		 */
		Map<String, Column> groupcolMetaMap = new HashMap<String, Column>(this.groupKeyfieldCount);

		groupKey = new UnsafeRow(this.groupKeyfieldCount);
		bufferHolder = new BufferHolder(groupKey, 0);
		unsafeRowWriter = new UnsafeRowWriter(bufferHolder, this.groupKeyfieldCount);
		bufferHolder.reset();

		Column curColMeta = null;

		for (int i = 0; i < this.groupKeyfieldCount; i++) {
			curColMeta = this.columToIndx.get(sortColumnsByIndex[i].toUpperCase());
			groupcolMetaMap.put(sortColumnsByIndex[i], curColMeta);

			switch (curColMeta.columType) {
			case Column.COL_TYPE_BIT:
				groupKey.setByte(i, (byte) 0);
				break;
			case Column.COL_TYPE_INT:
			case Column.COL_TYPE_INT24:
			case Column.COL_TYPE_LONG:
				groupKey.setInt(i, 0);
				break;
			case Column.COL_TYPE_SHORT:
				groupKey.setShort(i, (short) 0);
				break;
			case Column.COL_TYPE_FLOAT:
				groupKey.setFloat(i, 0);
				break;
			case Column.COL_TYPE_DOUBLE:
			case Column.COL_TYPE_NEWDECIMAL:
				groupKey.setDouble(i, 0);
				break;
			case Column.COL_TYPE_LONGLONG:
				groupKey.setLong(i, 0);
				break;
			default:
				unsafeRowWriter.write(i, "init".getBytes());
				break;
			}

		}
		groupKey.setTotalSize(bufferHolder.totalSize());

		groupKeySchema = new StructType(groupcolMetaMap, this.groupKeyfieldCount);
		groupKeySchema.setOrderCols(null);
	}

	private void initEmptyValueKey() {
		/**
		 * 构造valuerow
		 */
		emptyAggregationBuffer = new UnsafeRow(this.valuefieldCount);
		bufferHolder = new BufferHolder(emptyAggregationBuffer, 0);
		unsafeRowWriter = new UnsafeRowWriter(bufferHolder, this.valuefieldCount);
		bufferHolder.reset();

		Column curColMeta = null;
		for (Map.Entry<String, Column> fieldEntry : columToIndx.entrySet()) {
			curColMeta = fieldEntry.getValue();

			switch (curColMeta.columType) {
			case Column.COL_TYPE_BIT:
				emptyAggregationBuffer.setByte(curColMeta.columIndex, (byte) 0);
				break;
			case Column.COL_TYPE_INT:
			case Column.COL_TYPE_INT24:
			case Column.COL_TYPE_LONG:
				emptyAggregationBuffer.setInt(curColMeta.columIndex, 0);
				break;
			case Column.COL_TYPE_SHORT:
				emptyAggregationBuffer.setShort(curColMeta.columIndex, (short) 0);
				break;
			case Column.COL_TYPE_LONGLONG:
				emptyAggregationBuffer.setLong(curColMeta.columIndex, 0);
				break;
			case Column.COL_TYPE_FLOAT:
				emptyAggregationBuffer.setFloat(curColMeta.columIndex, 0);
				break;
			case Column.COL_TYPE_DOUBLE:
			case Column.COL_TYPE_NEWDECIMAL:
				emptyAggregationBuffer.setDouble(curColMeta.columIndex, 0);
				break;
			default:
				unsafeRowWriter.write(curColMeta.columIndex, "init".getBytes());
				break;
			}

		}

		emptyAggregationBuffer.setTotalSize(bufferHolder.totalSize());
		aggBufferSchema = new StructType(columToIndx, this.valuefieldCount);
		aggBufferSchema.setOrderCols(null);
	}

	public Iterator<UnsafeRow> getResult(UnsafeExternalRowSorter sorter) throws IOException {
		KVIterator<UnsafeRow, UnsafeRow> iter = aggregationMap.iterator();
		/**
		 * 求平均值
		 */
		if (isMergeAvg() && !isMergAvg) {
			try {
				while (iter.next()) {
					mergAvg(iter.getValue());
				}
			} catch (IOException e) {
				LOGGER.error(e.getMessage());
			}
			isMergAvg = true;
		}
		/**
		 * group having
		 */
		if (havingCols != null) {
			filterHaving(sorter);
		} else {

			/**
			 * KVIterator<K,V> ==>Iterator<V>
			 */
			insertValue(sorter);
		}
		return sorter.sort();
	}

	public UnsafeRow getAllBinaryRow(UnsafeRow unsafeRow) throws UnsupportedEncodingException {

		UnsafeRow value = new UnsafeRow(this.valuefieldCount);
		bufferHolder = new BufferHolder(value, 0);
		unsafeRowWriter = new UnsafeRowWriter(bufferHolder, this.valuefieldCount);
		bufferHolder.reset();
		Column colum = null;

		for (Map.Entry<String, Column> fieldEntry : columToIndx.entrySet()) {
			colum = fieldEntry.getValue();

			if (!unsafeRow.isNullAt(colum.columIndex)) {
				switch (colum.columType) {
				case Column.COL_TYPE_BIT:
					unsafeRowWriter.write(colum.columIndex, unsafeRow.getByte(colum.columIndex));
					break;
				case Column.COL_TYPE_INT:
				case Column.COL_TYPE_LONG:
				case Column.COL_TYPE_INT24:
					unsafeRowWriter.write(colum.columIndex, BytesTools.int2Bytes(unsafeRow.getInt(colum.columIndex)));
					break;
				case Column.COL_TYPE_SHORT:
					unsafeRowWriter.write(colum.columIndex, BytesTools.short2Bytes(unsafeRow.getShort(colum.columIndex)));
					break;
				case Column.COL_TYPE_LONGLONG:
					unsafeRowWriter.write(colum.columIndex, BytesTools.long2Bytes(unsafeRow.getLong(colum.columIndex)));
					break;
				case Column.COL_TYPE_FLOAT:
					unsafeRowWriter.write(colum.columIndex, BytesTools.float2Bytes(unsafeRow.getFloat(colum.columIndex)));
					break;
				case Column.COL_TYPE_DOUBLE:
				case Column.COL_TYPE_NEWDECIMAL:
					unsafeRowWriter.write(colum.columIndex, BytesTools.double2Bytes(unsafeRow.getDouble(colum.columIndex)));
					break;
				default:
					unsafeRowWriter.write(colum.columIndex, unsafeRow.getBinary(colum.columIndex));
					break;
				}
			} else {
				unsafeRowWriter.setNullAt(colum.columIndex);
			}
		}

		value.setTotalSize(bufferHolder.totalSize());
		return value;
	}

	private void insertValue(UnsafeExternalRowSorter sorter) {
		KVIterator<UnsafeRow, UnsafeRow> it = aggregationMap.iterator();
		try {
			while (it.next()) {
				UnsafeRow row = getAllBinaryRow(it.getValue());
				sorter.insertRow(row);
			}
		} catch (IOException e) {
			LOGGER.error(e.getMessage());
		}
	}

	private void filterHaving(UnsafeExternalRowSorter sorter) {

		if (havingCols.getColum() == null || aggregationMap == null) {
			return;
		}
		KVIterator<UnsafeRow, UnsafeRow> it = aggregationMap.iterator();
		byte[] right = havingCols.getRight().getBytes(StandardCharsets.UTF_8);
		int index = havingCols.getColum().getColumIndex();
		try {
			while (it.next()) {
				UnsafeRow row = getAllBinaryRow(it.getValue());
				switch (havingCols.getOperator()) {
				case "=":
					if (!eq(row.getBinary(index), right)) {
						sorter.insertRow(row);
					}
					break;
				case ">":
					if (!gt(row.getBinary(index), right)) {
						sorter.insertRow(row);
					}
					break;
				case "<":
					if (!lt(row.getBinary(index), right)) {
						sorter.insertRow(row);
					}
					break;
				case ">=":
					if (!gt(row.getBinary(index), right) && eq(row.getBinary(index), right)) {
						sorter.insertRow(row);
					}
					break;
				case "<=":
					if (!lt(row.getBinary(index), right) && eq(row.getBinary(index), right)) {
						sorter.insertRow(row);
					}
					break;
				case "!=":
					if (!neq(row.getBinary(index), right)) {
						sorter.insertRow(row);
					}
					break;
				}
			}
		} catch (IOException e) {
			LOGGER.error(e.getMessage());
		}

	}

	private boolean lt(byte[] l, byte[] r) {
		return -1 != ByteUtil.compareNumberByte(l, r);
	}

	private boolean gt(byte[] l, byte[] r) {
		return 1 != ByteUtil.compareNumberByte(l, r);
	}

	private boolean eq(byte[] l, byte[] r) {
		return 0 != ByteUtil.compareNumberByte(l, r);
	}

	private boolean neq(byte[] l, byte[] r) {
		return 0 == ByteUtil.compareNumberByte(l, r);
	}

	/**
	 * 构造groupKey
	 */
	private UnsafeRow getGroupKey(UnsafeRow row) throws UnsupportedEncodingException {

		UnsafeRow key = null;
		if (this.sortColumnsByIndex == null) {
			/**
			 * 针对没有group by关键字 select count(*) from table;
			 */
			key = new UnsafeRow(this.groupKeyfieldCount + 1);
			bufferHolder = new BufferHolder(key, 0);
			unsafeRowWriter = new UnsafeRowWriter(bufferHolder, this.groupKeyfieldCount + 1);
			bufferHolder.reset();
			unsafeRowWriter.write(0, "same".getBytes());
			key.setTotalSize(bufferHolder.totalSize());
			return key;
		}

		key = new UnsafeRow(this.groupKeyfieldCount);
		bufferHolder = new BufferHolder(key, 0);
		unsafeRowWriter = new UnsafeRowWriter(bufferHolder, this.groupKeyfieldCount);
		bufferHolder.reset();

		Column curColMeta = null;
		for (int i = 0; i < this.groupKeyfieldCount; i++) {
			curColMeta = this.columToIndx.get(sortColumnsByIndex[i].toUpperCase());
			if (!row.isNullAt(curColMeta.columIndex)) {
				switch (curColMeta.columType) {
				case Column.COL_TYPE_BIT:
					key.setByte(i, row.getByte(curColMeta.columIndex));
				case Column.COL_TYPE_INT:
				case Column.COL_TYPE_LONG:
				case Column.COL_TYPE_INT24:
					key.setInt(i, BytesTools.getInt(row.getBinary(curColMeta.columIndex)));
					break;
				case Column.COL_TYPE_SHORT:
					key.setShort(i, BytesTools.getShort(row.getBinary(curColMeta.columIndex)));
					break;
				case Column.COL_TYPE_FLOAT:
					key.setFloat(i, BytesTools.getFloat(row.getBinary(curColMeta.columIndex)));
					break;
				case Column.COL_TYPE_DOUBLE:
				case Column.COL_TYPE_NEWDECIMAL:
					key.setDouble(curColMeta.columIndex, BytesTools.getDouble(row.getBinary(curColMeta.columIndex)));
					break;
				case Column.COL_TYPE_LONGLONG:
					key.setLong(i, BytesTools.getLong(row.getBinary(curColMeta.columIndex)));
					break;
				default:
					unsafeRowWriter.write(i, row.getBinary(curColMeta.columIndex));
					break;
				}
			} else {
				key.setNullAt(i);
			}
		}

		key.setTotalSize(bufferHolder.totalSize());

		return key;
	}

	/**
	 * 构造value
	 */
	private UnsafeRow getValue(UnsafeRow row) throws UnsupportedEncodingException {

		UnsafeRow value = new UnsafeRow(this.valuefieldCount);
		bufferHolder = new BufferHolder(value, 0);
		unsafeRowWriter = new UnsafeRowWriter(bufferHolder, this.valuefieldCount);
		bufferHolder.reset();
		Column curColMeta = null;
		for (Map.Entry<String, Column> fieldEntry : columToIndx.entrySet()) {
			curColMeta = fieldEntry.getValue();
			if (!row.isNullAt(curColMeta.columIndex)) {
				switch (curColMeta.columType) {
				case Column.COL_TYPE_BIT:
					value.setByte(curColMeta.columIndex, row.getByte(curColMeta.columIndex));
					break;
				case Column.COL_TYPE_INT:
				case Column.COL_TYPE_LONG:
				case Column.COL_TYPE_INT24:
					value.setInt(curColMeta.columIndex, BytesTools.getInt(row.getBinary(curColMeta.columIndex)));

					LOGGER.error("getValue INT " + BytesTools.getInt(row.getBinary(curColMeta.columIndex)));
					break;
				case Column.COL_TYPE_SHORT:
					value.setShort(curColMeta.columIndex, BytesTools.getShort(row.getBinary(curColMeta.columIndex)));
					LOGGER.error("getValue toShort " + BytesTools.getShort(row.getBinary(curColMeta.columIndex)));
					break;
				case Column.COL_TYPE_LONGLONG:
					value.setLong(curColMeta.columIndex, BytesTools.getLong(row.getBinary(curColMeta.columIndex)));

					LOGGER.error("getValue COL_TYPE_LONGLONG ===> " + "size  " + row.getBinary(curColMeta.columIndex).length + "  " + ByteUtil.getLong(row.getBinary(curColMeta.columIndex)));

					break;
				case Column.COL_TYPE_FLOAT:
					value.setFloat(curColMeta.columIndex, BytesTools.getFloat(row.getBinary(curColMeta.columIndex)));
					LOGGER.error("getValue COL_TYPE_FLOAT " + BytesTools.getFloat(row.getBinary(curColMeta.columIndex)));

					break;
				case Column.COL_TYPE_DOUBLE:
				case Column.COL_TYPE_NEWDECIMAL:
					value.setDouble(curColMeta.columIndex, BytesTools.getDouble(row.getBinary(curColMeta.columIndex)));
					LOGGER.error("getValue COL_TYPE_NEWDECIMAL " + BytesTools.getDouble(row.getBinary(curColMeta.columIndex)));

					break;
				default:
					unsafeRowWriter.write(curColMeta.columIndex, row.getBinary(curColMeta.columIndex));
					break;
				}
			} else {
				value.setNullAt(curColMeta.columIndex);
				LOGGER.error("NULL");
			}
		}

		value.setTotalSize(bufferHolder.totalSize());
		return value;
	}

	public void addRow(UnsafeRow rowDataPkg) throws UnsupportedEncodingException {
		UnsafeRow key = getGroupKey(rowDataPkg);
		UnsafeRow value = getValue(rowDataPkg);

		if (aggregationMap.find(key)) {
			UnsafeRow rs = aggregationMap.getAggregationBuffer(key);
			aggregateRow(rs, value);
		} else {
			aggregationMap.put(key, value);
		}

		return;
	}

	private boolean isMergeAvg() {

		if (mergeColums == null) {
			return false;
		}

		for (AggregateFunctionType merg : mergeColums) {
			if (merg.mergeType == AggregateFunctionType.AVG) {
				return true;
			}
		}
		return false;
	}

	private void aggregateRow(UnsafeRow toRow, UnsafeRow newRow) throws UnsupportedEncodingException {
		if (mergeColums == null) {
			return;
		}

		for (AggregateFunctionType merg : mergeColums) {
			if (merg.mergeType != AggregateFunctionType.AVG) {
				byte[] result = null;
				byte[] left = null;
				byte[] right = null;
				int type = merg.colum.columType;
				int index = merg.colum.columIndex;
				switch (type) {
				case Column.COL_TYPE_INT:
				case Column.COL_TYPE_LONG:
				case Column.COL_TYPE_INT24:
					left = BytesTools.int2Bytes(toRow.getInt(index));
					right = BytesTools.int2Bytes(newRow.getInt(index));
					break;
				case Column.COL_TYPE_SHORT:
					left = BytesTools.short2Bytes(toRow.getShort(index));
					right = BytesTools.short2Bytes(newRow.getShort(index));
					break;
				case Column.COL_TYPE_LONGLONG:
					left = BytesTools.long2Bytes(toRow.getLong(index));
					right = BytesTools.long2Bytes(newRow.getLong(index));
					break;
				case Column.COL_TYPE_FLOAT:
					left = BytesTools.float2Bytes(toRow.getFloat(index));
					right = BytesTools.float2Bytes(newRow.getFloat(index));
					break;
				case Column.COL_TYPE_DOUBLE:
				case Column.COL_TYPE_NEWDECIMAL:
					left = BytesTools.double2Bytes(toRow.getDouble(index));
					right = BytesTools.double2Bytes(newRow.getDouble(index));
					break;
				default:
					break;
				}

				result = mertFields(left, right, type, merg.mergeType);

				if (result != null) {
					switch (type) {
					case Column.COL_TYPE_BIT:
						toRow.setByte(index, result[0]);
					case Column.COL_TYPE_INT:
					case Column.COL_TYPE_LONG:
					case Column.COL_TYPE_INT24:
						toRow.setInt(index, BytesTools.getInt(result));
						break;
					case Column.COL_TYPE_SHORT:
						toRow.setShort(index, BytesTools.getShort(result));
						break;
					case Column.COL_TYPE_LONGLONG:
						toRow.setLong(index, BytesTools.getLong(result));
						break;
					case Column.COL_TYPE_FLOAT:
						toRow.setFloat(index, BytesTools.getFloat(result));
						break;
					case Column.COL_TYPE_DOUBLE:
					case Column.COL_TYPE_NEWDECIMAL:
						toRow.setDouble(index, BytesTools.getDouble(result));
						break;
					default:
						break;
					}
				}
			}
		}
	}

	private void mergAvg(UnsafeRow toRow) throws UnsupportedEncodingException {

		if (mergeColums == null) {
			return;
		}

		for (AggregateFunctionType merg : mergeColums) {
			if (merg.mergeType == AggregateFunctionType.AVG) {
				byte[] result = null;
				byte[] avgSum = null;
				byte[] avgCount = null;

				int type = merg.colum.columType;
				int avgSumIndex = merg.colum.avgSumIndex;
				int avgCountIndex = merg.colum.avgCountIndex;

				switch (type) {
				case Column.COL_TYPE_BIT:
					avgSum = BytesTools.toBytes(toRow.getByte(avgSumIndex));
					avgCount = BytesTools.toBytes(toRow.getLong(avgCountIndex));
					break;
				case Column.COL_TYPE_INT:
				case Column.COL_TYPE_LONG:
				case Column.COL_TYPE_INT24:
					avgSum = BytesTools.int2Bytes(toRow.getInt(avgSumIndex));
					avgCount = BytesTools.long2Bytes(toRow.getLong(avgCountIndex));
					break;
				case Column.COL_TYPE_SHORT:
					avgSum = BytesTools.short2Bytes(toRow.getShort(avgSumIndex));
					avgCount = BytesTools.long2Bytes(toRow.getLong(avgCountIndex));
					break;

				case Column.COL_TYPE_LONGLONG:
					avgSum = BytesTools.long2Bytes(toRow.getLong(avgSumIndex));
					avgCount = BytesTools.long2Bytes(toRow.getLong(avgCountIndex));

					break;
				case Column.COL_TYPE_FLOAT:
					avgSum = BytesTools.float2Bytes(toRow.getFloat(avgSumIndex));
					avgCount = BytesTools.long2Bytes(toRow.getLong(avgCountIndex));

					break;
				case Column.COL_TYPE_DOUBLE:
				case Column.COL_TYPE_NEWDECIMAL:
					avgSum = BytesTools.double2Bytes(toRow.getDouble(avgSumIndex));
					avgCount = BytesTools.long2Bytes(toRow.getLong(avgCountIndex));
					break;
				default:
					break;
				}

				result = mertFields(avgSum, avgCount, merg.colum.columType, merg.mergeType);

				if (result != null) {
					switch (type) {
					case Column.COL_TYPE_BIT:
						toRow.setByte(avgSumIndex, result[0]);
						break;
					case Column.COL_TYPE_INT:
					case Column.COL_TYPE_LONG:
					case Column.COL_TYPE_INT24:
						toRow.setInt(avgSumIndex, BytesTools.getInt(result));
						break;
					case Column.COL_TYPE_SHORT:
						toRow.setShort(avgSumIndex, BytesTools.getShort(result));
						break;
					case Column.COL_TYPE_LONGLONG:
						toRow.setLong(avgSumIndex, BytesTools.getLong(result));
						break;
					case Column.COL_TYPE_FLOAT:
						toRow.setFloat(avgSumIndex, BytesTools.getFloat(result));
						break;
					case Column.COL_TYPE_DOUBLE:
					case Column.COL_TYPE_NEWDECIMAL:
						toRow.setDouble(avgSumIndex, ByteUtil.getDouble(result));
						break;
					default:
						break;
					}
				}
			}
		}
	}

	private byte[] mertFields(byte[] bs, byte[] bs2, int colType, int mergeType) throws UnsupportedEncodingException {

		if (bs2 == null || bs2.length == 0) {
			return bs;
		} else if (bs == null || bs.length == 0) {
			return bs2;
		}

		switch (mergeType) {
		case AggregateFunctionType.SUM:
			if (colType == Column.COL_TYPE_NEWDECIMAL || colType == Column.COL_TYPE_DECIMAL || colType == Column.COL_TYPE_DOUBLE || colType == Column.COL_TYPE_FLOAT) {
				double vale = BytesTools.getDouble(bs) + BytesTools.getDouble(bs2);
				LOGGER.error("MERGE_SUM " + vale);
				return BytesTools.double2Bytes(vale);
			}

		case AggregateFunctionType.COUNT: {
			long s1 = BytesTools.getLong(bs);
			long s2 = BytesTools.getLong(bs2);
			long total = s1 + s2;
			LOGGER.error("total " + total);
			return BytesTools.long2Bytes(total);
		}

		case AggregateFunctionType.MAX: {
			int compare = BytesTools.compareTo(bs, bs2);
			return (compare > 0) ? bs : bs2;
		}

		case AggregateFunctionType.MIN: {
			int compare = BytesTools.compareTo(bs, bs2);
			return (compare > 0) ? bs2 : bs;

		}
		case AggregateFunctionType.AVG: {
			/**
			 * 数值总和
			 */
			double sum = BytesTools.getDouble(bs);

			/**
			 * 元素总个数
			 */
			long count = BytesTools.getLong(bs2);
			double value = sum / count;
			NumberFormat nf = NumberFormat.getNumberInstance();
			nf.setMaximumFractionDigits(4);
			LOGGER.error("MERGE_SUM " + value);
			return BytesTools.double2Bytes(value);
		}
		default:
			return null;
		}
	}

	public void free() {
		if (aggregationMap != null)
			aggregationMap.free();
	}
}
